#include "Rocksdb.h"

#include <cassert>
#include "rocksdb/db.h"
#include "OptionsArgs.h"

#include <iostream>


namespace helper {
namespace demo {

class RocksdbIterator : public Iterator<Slice, Slice>{
public:
  using super = Iterator<Slice, Slice>;
  using key_type = super::key_type;
  using value_type = super::value_type;
  using pair_type = super::pair_type;

  RocksdbIterator(::rocksdb::DB * db, const ::rocksdb::Comparator * comp, const ReadOptions & read, const key_type & start, const key_type & end);
  ~RocksdbIterator() override ;

  pair_type current() const override;

  void next() override ;
  bool hasNext() const override ;

private:
  ::rocksdb::Iterator * impl_;
  const ::rocksdb::Comparator * comp_;
  key_type start_;
  const key_type end_;
  mutable key_type key_;
  mutable value_type value_;
};

RocksdbIterator::RocksdbIterator(::rocksdb::DB * db, const ::rocksdb::Comparator * comp, const ReadOptions & read, const key_type & start, const key_type & end)
  : impl_(db->NewIterator(read))
  , comp_(comp)
  , start_(start) , end_(end)
{
  if(start_ == key_type()){
    impl_->SeekToFirst();
  }
  else{
    impl_->Seek(start_);
  }
}
RocksdbIterator::~RocksdbIterator() {
  if(impl_ != nullptr){
    delete impl_;
    impl_ = nullptr;
  }
}

RocksdbIterator::pair_type RocksdbIterator::current() const {
  key_ = impl_->key();
  value_ = impl_->value();
  return pair_type(key_, value_);
};

void RocksdbIterator::next() {
  impl_->Next();
}
bool RocksdbIterator::hasNext() const {
  assert(comp_ != nullptr);
  if(end_.size() == 0)
    return impl_->Valid();
  return impl_->Valid() && comp_->Compare(impl_->key(), end_) < 0;
}




Rocksdb * Rocksdb::Open(const std::string & name, const OptionsArgs & optionsArgs){
  Rocksdb * db = new Rocksdb(name, optionsArgs);
  if(!db->open()){
    delete db;
    return nullptr;
  }
  return db;
}
bool Rocksdb::destroy(const std::string & name, const Options & options) {
  //easy impl
  return ::rocksdb::DestroyDB(name, options).ok();
}

Rocksdb::Rocksdb(const std::string & name, const OptionsArgs & optionsArgs) 
  :name_(name)
	, optionsArgs_(new OptionsArgs(optionsArgs))
	, impl_(nullptr)
{
}

Rocksdb::~Rocksdb() {
  this->close();
	if(impl_ != nullptr){
		delete impl_;
		impl_ = nullptr;
	}
}

IDBInstance::DBImpl * Rocksdb::instance() {
  return reinterpret_cast<DBImpl*>(impl_);
}

const std::string & Rocksdb::name() const {
  return name_;
}

Status Rocksdb::put(const Slice & key, const Slice & value) {
  assert(impl_ != nullptr);
  return impl_->Put(optionsArgs_->write(), key, value);
}

Status Rocksdb::get(const Slice & key, std::string & value) const {
  assert(impl_ != nullptr);
  return impl_->Get(optionsArgs_->read(), key, &value);
}
Status Rocksdb::del(const Slice & key) {
  assert(impl_ != nullptr);
  return impl_->Delete(optionsArgs_->write(), key);
}

bool Rocksdb::foreach(const visitor & v, const Slice & start, const Slice & end) const {
  assert(impl_ != nullptr);
  return this->foreach(optionsArgs_->read(), v, start, end);
}
Rocksdb::iterator * Rocksdb::range(const key_type & start, const key_type & end) const {
  assert(impl_ != nullptr);
  return this->range(optionsArgs_->read(), start, end);
}

Status Rocksdb::put(const WriteOptions & write, const Slice & key, const Slice & value) {
  assert(impl_ != nullptr);
  return impl_->Put(write, key, value);
}
Status Rocksdb::get(const ReadOptions & read, const Slice & key, std::string & value) const {
  assert(impl_ != nullptr);
  return impl_->Get(read, key, &value);
}
Status Rocksdb::del(const WriteOptions & write, const Slice & key) {
  assert(impl_ != nullptr);
  return impl_->Delete(write, key);
}
bool Rocksdb::foreach(const ReadOptions & read, const visitor & v, const Slice & start, const Slice & end) const {
  assert(impl_ != nullptr);
  std::unique_ptr<::rocksdb::Iterator> iter(impl_->NewIterator(read));
  if(iter == nullptr) 
    return false;
  
  if(start.size() == 0){
    iter->SeekToFirst();
  }
  else{
    iter->Seek(start);
  }

  if(end.size() == 0){ // end == Slice()
    for(; iter->Valid(); iter->Next()){
      if(!v(iter->key(), iter->value()))  return false;
    }
  }
  else{ // end is RealKey
    auto * comparator = optionsArgs_->columnFamily().comparator;
    assert(comparator != nullptr);
    for(; iter->Valid() && comparator->Compare(iter->key(), end) < 0 ; iter->Next()){
      if(!v(iter->key(), iter->value()))  return false;
    }
  }
  return true;
}

Rocksdb::iterator * Rocksdb::range(const ReadOptions & read, const key_type & start, const key_type & end) const {
  assert(impl_ != nullptr);
  return new RocksdbIterator(impl_, optionsArgs_->columnFamily().comparator, read, start, end);
}

const OptionsArgs & Rocksdb::optionsArgs() const {
  return *optionsArgs_;
}
OptionsArgs & Rocksdb::optionsArgs() { //changeable
  return *optionsArgs_;
}

bool Rocksdb::_open() {
#if 0
  bool ret = ::rocksdb::DB::Open(optionsArgs_->open(), name_, &impl_).ok();
  std::cout << name_ << ":.open..ret:" << ret << std::endl;
  return ret;
#else

  return ::rocksdb::DB::Open(optionsArgs_->open(), name_, &impl_).ok();
	//assert(impl != nullptr); here
#endif
};
bool Rocksdb::_close()  {
  assert(impl_ != nullptr);

  return impl_->Close().ok();
};



}//namespace demo
}//namespace helper
